package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo16Checkpoint {
  def main(args: Array[String]): Unit = {
    //配置spark文件
    val conf: SparkConf = new SparkConf().setAppName("Checkpoint").setMaster("local")

    //配置spark对象
    val sc = new SparkContext(conf)

    //读取students数据
    val students: RDD[String] = sc.textFile("data/students.txt")

    //指定checkpoint保存的路径
    /**
      * Checkpoint 的产生就是为了相对而言更加可靠的持久化数据，
      * 在 Checkpoint 可以指定把数据放在本地并且是多副本的方式，
      * 在正常生产环境下通常放在 HDFS 上，借助HDFS 高可靠的特征来实现更可靠的数据持久化。
      *
      */
    sc.setCheckpointDir("data/checkpoint")

    //切分获取数据
    val studentsRDD: RDD[(String, String, Int, String, String)] = students.map(stu => {

      println("studentRDD处理")
      //切分数据
      val split: Array[String] = stu.split(",")
      val id: String = split(0)
      val name: String = split(1)
      val age: Int = split(2).toInt
      val gender: String = split(3)
      val clazz: String = split(4)
      (id,name,age,gender,clazz)
    })
    studentsRDD.cache()
    studentsRDD.checkpoint()
    /**
      * checkpoint流程
      * 1、当第一个job执行完成之后会从后面回溯，如果rdd调用了checkpoint方法会被打上一个标记
      * 2、另启动一个job重新计算这个rdd的数据，将rdd的数据保存到hdfs
      * 3、后续的job 就可以直接使用checkpoint中的数据
      *
      * 优化：可以在checkpoint执行cache
      *
      * checkpoint主要用在spark steaming
      *
      */
    //统计每个班级人数
    studentsRDD.map(stu => (stu._5,1)).reduceByKey(_+_).foreach(println)


    //统计男女生性别人数
    studentsRDD.map(stu => (stu._4,1)).reduceByKey(_+_).foreach(println)
  }
}
